package com.distributer.delayqueue.autoconfigure;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import com.alibaba.fastjson.JSONObject;
import com.distributer.delayqueue.delayqueue.JobOperationService;
import com.distributer.delayqueue.delayqueue.Queue;
import com.distributer.delayqueue.delayqueue.bucket.BucketQueueManager;
import com.distributer.delayqueue.delayqueue.event.EventBus;
import com.distributer.delayqueue.delayqueue.event.RdbOperation;
import com.distributer.delayqueue.delayqueue.event.RedisJobTraceEvent;
import com.distributer.delayqueue.delayqueue.ready.ReadyQueueManager;
import com.distributer.delayqueue.delayqueue.support.DistributedLock;
import com.distributer.delayqueue.delayqueue.support.RedisQueueProperties;
import com.distributer.delayqueue.job.JobMsg;
import com.distributer.delayqueue.job.JobStatus;
import com.distributer.delayqueue.job.JobWrapp;
import com.distributer.delayqueue.util.NamedUtil;
	

class RedisQueueImpl implements Queue,java.io.Closeable{

	private static final Logger LOGGER = LoggerFactory.getLogger(RedisQueueImpl.class);
	
	private static final AtomicBoolean isRunning = new AtomicBoolean(false);
	private JobOperationService jobOperationService;
	private RedisQueueProperties redisQueueProperties;
	private ReadyQueueManager readyQueueManager;
	private BucketQueueManager bucketQueueManager;
	private DistributedLock lock;
	
	@Override
	public void putJobMsg(JobMsg jobMsg) {
		try {
			Assert.notNull(jobMsg, "job为空！");
			Assert.notNull(jobMsg.getJobId(), "jobId为空！");
			Assert.notNull(jobMsg.getDelayTime(), "jobDelayTime为空！");
			if (jobMsg.getStatus() != JobStatus.WaitPut.ordinal() && jobMsg.getStatus() != JobStatus.Restore.ordinal()) {
				throw new IllegalArgumentException(String.format("jobId:%s,状态异常！",jobMsg.getJobId()));
			}
			//新增job到元数据池中
			this.jobOperationService.addJobToPool(jobMsg);
			EventBus.getInstance().postEvent(new RedisJobTraceEvent(jobMsg, RdbOperation.INSERT));
			//添加job到延时队列中
			String bucketName = NamedUtil.buildBucketName(redisQueueProperties.getProfix(), redisQueueProperties.getName(), bucketIndex(jobMsg.getJobId(),redisQueueProperties.getBucketSize()));
			if (jobMsg instanceof JobWrapp) {
				((JobWrapp)jobMsg).setBucketName(bucketName);
			}
			long score = Long.valueOf(jobMsg.getCreateTime().getTime() + jobMsg.getDelayTime());
			SimpleDateFormat dateFormat = new SimpleDateFormat("YYYY-MM-DD mm:hh:ss");
			//LOGGER.info(String.format("添加job：%s 到延时队列：%s 入队时间： %s 延时时间为：%s", jobMsg.getJobId(),bucketName,dateFormat.format(new Date()),score));
			this.jobOperationService.addBucketJob(bucketName, jobMsg.getJobId(), score);
			this.jobOperationService.updateJobStatus(jobMsg.getJobId(), JobStatus.Delay);
			//持久化到数据库
			jobMsg.setStatus(JobStatus.Delay.ordinal());
			EventBus.getInstance().postEvent(new RedisJobTraceEvent(jobMsg));
			//LOGGER.info("添加job成功，jobId:{}",jobMsg.getJobId());
		} catch (Exception e) {
			LOGGER.error("添加失败,job信息：{}",JSONObject.toJSONString(jobMsg),e);
		}
	}

	@Override
	public boolean deleteJob(String jobId) {
		boolean result = false;
		JobWrapp jobMsg = (JobWrapp) getJobByJobId(jobId);
		if (jobMsg == null) {
			LOGGER.info("根据jobId未找到job信息！");
			return false;
		}
		try{
			if (lock.lock(jobId)) {
				if (jobMsg.getStatus() == JobStatus.Finish.ordinal()) {
					LOGGER.info("jobId的任务已经执行完成！");
					return false;
				}
				if (!StringUtils.isEmpty(jobMsg.getBucketName())) {
					this.jobOperationService.removeBucketKey(jobMsg.getBucketName(), jobId);
					if (jobMsg.getStatus()==JobStatus.Ready.ordinal()) {
						this.jobOperationService.removeReadyJob(jobId);
					}
				}
				jobMsg.setStatus(JobStatus.Delete.ordinal());
				this.jobOperationService.removeJobToPool(jobId);
				EventBus.getInstance().postEvent(new RedisJobTraceEvent(jobMsg));
				result = true;
			}
			return result;
		}finally{
			lock.unlock(jobId);
		}
	}

	@Override
	public void ack(String jobId) {
		
	}

	@Override
	public JobMsg getJobByJobId(String jobId) {
		if (StringUtils.isEmpty(jobId)) {
			throw new NullPointerException("jobId为空！");
		}
		JobMsg jobMsg = this.jobOperationService.getJob(jobId);
		return jobMsg;
	}

	@Override
	public long getSize() {
		return 0;
	}

	@Override
	public void clear() {
		
	}
	
	private static int bucketIndex(String jobId,int bucketSize){
		Random random = new Random();
		return random.nextInt(bucketSize);
	}
	
	@Override
	public void start() {
		if (isRunning.compareAndSet(false, true)) {
			bucketQueueManager.start();
			readyQueueManager.start();
			LOGGER.info("started queue....");
		}
	}

	@Override
	public void stop() {
		if (isRunning.compareAndSet(true, false)) {
			bucketQueueManager.stop();
			readyQueueManager.stop();
			LOGGER.info("stop queue...");
		}
	}

	@Override
	public boolean isRunning() {
		return isRunning.get();
	}

	@Override
	public void close() throws IOException {
		stop();
	}

	public void setJobOperationService(JobOperationService jobOperationService) {
		this.jobOperationService = jobOperationService;
	}

	public void setRedisQueueProperties(RedisQueueProperties redisQueueProperties) {
		this.redisQueueProperties = redisQueueProperties;
	}

	public void setReadyQueueManager(ReadyQueueManager readyQueueManager) {
		this.readyQueueManager = readyQueueManager;
	}

	public void setBucketQueueManager(BucketQueueManager bucketQueueManager) {
		this.bucketQueueManager = bucketQueueManager;
	}

	public void setLock(DistributedLock lock) {
		this.lock = lock;
	}
	
}